package main

import (
	"fmt"
	"log"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	inf "github.com/fzdwx/infinite"
	"github.com/fzdwx/infinite/components/input/text"
	"github.com/fzdwx/infinite/theme"
)

// 节点信息
type BrokerInfo struct {
	// 节点Ip:端口
	Servers []string
	// 消息主题
	Topic string
	// 分区
	Partition int32
	// 消费类型
	OffsetType int64
}

var broker = &BrokerInfo{}

func createText(tip, defaultVal string) *text.Text {
	return inf.NewText(
		text.WithRequired(),
		text.WithPrompt(tip),
		text.WithPromptStyle(theme.DefaultTheme.PromptStyle),
		text.WithDefaultValue(defaultVal),
	)
}

func main() {
	title := `
	_          __ _                                                         
	| |        / _| |                                                        
	| | ____ _| |_| | ____ _     ___ ___  _ __  ___ _   _ _ __ ___   ___ _ 
	| |/ / _  |  _| |/ / _ |   / __/ _ \| '_ \/ __| | | | '_  _ \ / _ \ __|
	|   < (_| | | |   < (_| | | (_| (_) | | | \__ \ |_| | | | | | |  __/ |   
	|_|\_\__,_|_| |_|\_\__,_|  \___\___/|_| |_|___/\__,_|_| |_| |_|\___|_|   
	
	                             exit/quit 
	`
	fmt.Println(title)
	tip("servers", "please input ip and port (eg: host1:9092,host2:9092)", 0, handleServers)
	tip("topic", "please input topic", 0, handleTopic)
	tip("partition", "please input partition(eg: 0)", 0, handlePartition)
	tip("offset type", "please input offset type: 0(earliest)/1(latest)?", 0, handleOffset)

	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true

	master, err := sarama.NewConsumer(broker.Servers, config)
	if err != nil {
		log.Fatalf("connect error: %#v", err.Error())
		log.Panic(err)
	}
	defer func() {
		closeConsumer(master)
	}()

	consumer, err := master.ConsumePartition(broker.Topic, broker.Partition, broker.OffsetType)
	if err != nil {
		log.Panic(err)
	}
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)
	doneCh := make(chan struct{})
	go func() {
		for {
			select {
			case err := <-consumer.Errors():
				log.Println(err)
			case msg := <-consumer.Messages():
				log.Println("received message:", string(msg.Key), string(msg.Value))
			case <-signals:
				log.Println("interrupt is detected")
				doneCh <- struct{}{}
			}
		}
	}()
	<-doneCh
}

func handleServers(servers string) error {
	broker.Servers = strings.Split(servers, ",")
	return nil
}

func handleTopic(topic string) error {
	broker.Topic = topic
	return nil
}

func handlePartition(partition string) error {
	part, err := strconv.Atoi(partition)
	if err != nil {
		return err
	}

	broker.Partition = int32(part)
	return nil
}

func handleOffset(offsetType string) error {
	switch offsetType {
	case "0":
		broker.OffsetType = sarama.OffsetOldest
	case "1":
		broker.OffsetType = sarama.OffsetNewest
	}
	return nil
}

func closeConsumer(consumer sarama.Consumer) {
	log.Println("close consumer")
	if err := consumer.Close(); err != nil {
		log.Panic(err)
	}
}

func tip(message, defaultVal string, count int, handleFun func(value string) error) {
	if count > 2 {
		fmt.Printf("too many errors(%d times), system will exit after 3 second.\n", count)
		time.Sleep(time.Second * 2)
		os.Exit(0)
	}
	if count > 0 {
		fmt.Printf("input error(%d times), retry please……\n", count)
	}
	value, err := createText(message, defaultVal).Display()
	if err != nil {
		tip(message, defaultVal, count+1, handleFun)
	}
	switch value {
	case "exit", "quit":
		os.Exit(0)
	}
	if err = handleFun(value); err != nil {
		tip(message, defaultVal, count+1, handleFun)
	}
}
